package cn.followtry.zk;

/**
 * brief-zk/cn.followtry.zk.DataMonitor
 *
 * @since 2017年1月5日 下午3:11:50
 */

import java.util.Arrays;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 * A simple class that monitors the data and existence of a ZooKeeper node. It uses asynchronous
 * ZooKeeper APIs.
 */
public class DataMonitor implements Watcher, StatCallback {

  private ZooKeeper zk;

  private String znode;

  private Watcher chainedWatcher;

  private boolean dead;

  private DataMonitorListener listener;

  private byte[] prevData;

  /**
   * .
   */
  public DataMonitor(ZooKeeper zk,String znode,Watcher chainedWatcher,DataMonitorListener
          listener) {
    this.zk = zk;
    this.znode = znode;
    this.chainedWatcher = chainedWatcher;
    this.listener = listener;
    // Get things started by checking if the node exists. We are going
    // to be completely event driven
    zk.exists(znode,true,this,null);
  }

  /**
   * Other classes use the DataMonitor by implementing this method.
   */
  public interface DataMonitorListener {
    /**
     * The existence status of the node has changed.
     */
    void exists(byte[] data);

    /**
     * The ZooKeeper session is no longer valid.
     *
     * @param rc the ZooKeeper reason code
     */
    void closing(int rc);
  }

  /**
   * .
   */
  public void process(WatchedEvent event) {
    String path = event.getPath();
    if (event.getType() == Event.EventType.None) {
      // We are are being told that the state of the
      // connection has changed
      switch (event.getState()) {
        case SyncConnected:
          // In this particular example we don't need to do anything
          // here - watches are automatically re-registered with
          // server and any watches triggered while the client was
          // disconnected will be delivered (in order of course)
          break;
        case Expired:
          // It's all over
          dead = true;
          listener.closing(KeeperException.Code.SESSIONEXPIRED.intValue());
          break;
        default:
          break;
      }
    } else {
      if (path != null && path.equals(znode)) {
        // Something has changed on the node, let's find out
        zk.exists(znode,true,this,null);
      }
    }
    if (chainedWatcher != null) {
      chainedWatcher.process(event);
    }
  }

  /**
   * .
   */
  public void processResult(int rc,String path,Object ctx,Stat stat) {
    boolean exists;

    if (rc == Code.OK.intValue()) {
      exists = true;
    } else if (rc == Code.NONODE.intValue()) {
      exists = false;
    } else if (rc == Code.SESSIONEXPIRED.intValue() || rc == Code.NOAUTH.intValue()) {
      dead = true;
      listener.closing(rc);
      return;
    } else {
      // Retry errors
      zk.exists(znode,true,this,null);
      return;
    }

    byte[] b = null;
    if (exists) {
      try {
        b = zk.getData(znode,false,null);
      } catch (KeeperException e) {
        // We don't need to worry about recovering now. The watch
        // callbacks will kick off any exception handling
        e.printStackTrace();
      } catch (InterruptedException e) {
        return;
      }
    }
    if ((b == null && b != prevData) || (b != null && !Arrays.equals(prevData,b))) {
      listener.exists(b);
      prevData = b;
    }
  }
}
